package com.zhang.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhang.utils.DateFormatUtil;
import com.zhang.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 日志分流
 * @author: zhang
 * @date: 2022/5/5 16:02
 * 数据流：web/app -> nginx -> 日志服务器（log） -> Flume -> Kafka(ods) -> FlinkApp -> Kafka(dwd)
 * 程序：Mock -> f1.sh -> Kafka(zk) -> BaseLogApp -> Kafka(ZK)
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //TODO 读取kafka topic_log 主题数据创建流
        //TODO 将数据转换为JSON格式，并过滤非JSON格式数据

        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };

        OutputTag<String> dirtyTag = new OutputTag<String>("Dirty") {
        };
        SingleOutputStreamOperator<JSONObject> kafkaDS = env


                .addSource(MyKafkaUtil.getKafkaConsumer("topic_log", "base_log_app"))
                .process(new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(value);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, value);
                        }
                    }
                });

        //TODO 使用状态编程做新老用户校验
        //TODO 使用侧输出流对数据进行分流处理
        //TODO 页面浏览：主流
        //TODO 启动日志：侧输出流
        //TODO 曝光日志：侧输出流
        //TODO 动作日志：侧输出流
        //TODO 错误日志：侧输出流
        SingleOutputStreamOperator<String> pageDS = kafkaDS
                .keyBy(data -> data.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {
                    private ValueState<String> stateTime;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        stateTime = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>(
                                        "state-time", Types.STRING
                                )
                        );
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        String isNew = value.getJSONObject("common").getString("is_new");
                        String lastTime = stateTime.value();
                        Long ts = value.getLong("ts");
                        if (isNew.equals("1")) {
                            // 获取当前数据时间
                            String currTime = DateFormatUtil.toDate(ts);
                            if (lastTime == null) {
                                stateTime.update(currTime);
                            } else if (!currTime.equals(lastTime)) {
                                value.getJSONObject("common").put("is_new", "0");
                            }
                        } else if (lastTime == null) {
                            String yesterday = DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L);
                            stateTime.update(yesterday);
                        }

                        return value;
                    }
                })
                .process(new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        String jsonString = value.toJSONString();
                        // 尝试取出数据error字段
                        String error = value.getString("err");
                        if (error != null) {
                            ctx.output(errorTag, jsonString);
                        }
                        // 尝试获取启动字段
                        String start = value.getString("start");
                        if (start != null) {
                            ctx.output(startTag, jsonString);
                        } else {

                            String pageId = value.getJSONObject("page").getString("page_id");
                            Long ts = value.getLong("ts");
                            JSONObject common = value.getJSONObject("common");
                            // 尝试获取曝光数组
                            JSONArray displaysArray = value.getJSONArray("displays");
                            if (displaysArray != null && displaysArray.size() > 0) {
                                for (int i = 0; i < displaysArray.size(); i++) {
                                    JSONObject displaysJson = displaysArray.getJSONObject(i);
                                    displaysJson.put("ts", ts);
                                    displaysJson.put("page_id", pageId);
                                    displaysJson.put("common", common);
                                    ctx.output(displayTag, displaysJson.toJSONString());
                                }
                            }

                            // 尝试获取动作数组
                            JSONArray actionsArray = value.getJSONArray("actions");
                            if (actionsArray != null && actionsArray.size() > 0) {
                                for (int i = 0; i < actionsArray.size(); i++) {
                                    JSONObject actionsJson = actionsArray.getJSONObject(i);
                                    actionsJson.put("ts", ts);
                                    actionsJson.put("page_id", pageId);
                                    actionsJson.put("common", common);
                                    ctx.output(actionTag, actionsJson.toJSONString());
                                }
                            }

                            // 页面日志输出到主流
                            value.remove("displays");
                            value.remove("actions");
                            out.collect(value.toJSONString());
                        }
                    }
                });

        //TODO 提取各个流的数据
        pageDS.print("page>>>>>");

        kafkaDS.getSideOutput(dirtyTag).print("dirty>>>>>>");
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);

        startDS.print("start>>>>");
        errorDS.print("error>>>>");
        displayDS.print("display>>>>");
        actionDS.print("action>>>>");

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        pageDS.addSink(MyKafkaUtil.getKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getKafkaProducer(start_topic));
        errorDS.addSink(MyKafkaUtil.getKafkaProducer(error_topic));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getKafkaProducer(action_topic));
        //TODO 执行任务
        env.execute();
    }
}
